package com.xiongmm.stream.support;

import com.google.common.base.Function;
import com.google.common.collect.Ordering;
import com.google.common.primitives.Ints;
import com.ucar.opentsdb.client.value.request.Point;
import com.xiongmm.stream.domain.InstanceInfoAllVo;
import com.xiongmm.stream.domain.InstanceInfoCmdVo;
import com.xiongmm.stream.utils.BambooUtils;
import org.apache.commons.collections.MapUtils;
import org.apache.flink.shaded.curator.org.apache.curator.shaded.com.google.common.base.CharMatcher;
import org.apache.flink.shaded.curator.org.apache.curator.shaded.com.google.common.base.Splitter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Nullable;
import java.util.*;

import static com.xiongmm.stream.support.OpentsdbConstant.*;

/**
 * @author: XiongMM
 * @date: 2019/6/21 15:52
 * @description:
 */
public class FlinkSupportor {

    private final static Logger LOGGER = LoggerFactory.getLogger(FlinkSupportor.class);

    /**
     * 入库-实例监控大盘
     *
     * @param infoAllVo
     * @param metricInstanceMonitor
     * @param unixTime
     */
    public static List<Point> processInfoInstanceMetric(InstanceInfoAllVo infoAllVo, String metricInstanceMonitor, long unixTime) {
        List<Point> list = new ArrayList<>();
        try {
            Map<String, String> tags = new HashMap<>(3);
            tags.put(OpentsdbConstant.COMMON_TAG_APPID, String.valueOf(infoAllVo.getAppId()));
            tags.put(OpentsdbConstant.COMMON_TAG_INSTANCE, infoAllVo.getIp() + "_" + infoAllVo.getPort());
            //入库角色
            tags.put(OpentsdbConstant.COMMON_TAG_INDICARTOR, INDICATOR_ROLE);
            Point point = Point.metric(metricInstanceMonitor).tag(tags).value(unixTime, infoAllVo.getRole()).build(true);
            list.add(point);
            //入库fork耗时
            tags.put(OpentsdbConstant.COMMON_TAG_INDICARTOR, INDICATOR_LATEST_FORK_USEC);
            point = Point.metric(metricInstanceMonitor).tag(tags).value(unixTime, infoAllVo.getLatestForkUsec()).build(true);
            list.add(point);
            //入库是否进行AOF重写
            tags.put(OpentsdbConstant.COMMON_TAG_INDICARTOR, INDICATOR_AOF_REWRITE_IN_PROGRESS);
            point = Point.metric(metricInstanceMonitor).tag(tags).value(unixTime, infoAllVo.getAofRewriteInProgress()).build(true);
            list.add(point);
            //入库创建aof文件耗费时长
            tags.put(OpentsdbConstant.COMMON_TAG_INDICARTOR, INDICATOR_AOF_LAST_REWRITE_TIME_SEC);
            point = Point.metric(metricInstanceMonitor).tag(tags).value(unixTime, infoAllVo.getAofLastRewriteTimeSec()).build(true);
            list.add(point);
            //入库当前AOF文件大小
            tags.put(OpentsdbConstant.COMMON_TAG_INDICARTOR, INDICATOR_AOF_CURRENT_SIZE);
            point = Point.metric(metricInstanceMonitor).tag(tags).value(unixTime, infoAllVo.getAofCurrentSize()).build(true);
            list.add(point);
            //入库AOF上次启动或rewrite的大小
            tags.put(OpentsdbConstant.COMMON_TAG_INDICARTOR, INDICATOR_AOF_BASE_SIZE);
            point = Point.metric(metricInstanceMonitor).tag(tags).value(unixTime, infoAllVo.getAofBaseSize()).build(true);
            list.add(point);
            //入库客户端输入缓冲区大小
            tags.put(OpentsdbConstant.COMMON_TAG_INDICARTOR, INDICATOR_CLIENT_INPUT_BUF);
            point = Point.metric(metricInstanceMonitor).tag(tags).value(unixTime, infoAllVo.getClientInputBuf()).build(true);
            list.add(point);
            //入库客户端输出缓冲区列表
            tags.put(OpentsdbConstant.COMMON_TAG_INDICARTOR, INDICATOR_CLIENT_OUTPUT_LIST);
            point = Point.metric(metricInstanceMonitor).tag(tags).value(unixTime, infoAllVo.getClientOutputList()).build(true);
            list.add(point);
            //上次bgsave是否成功
            tags.put(OpentsdbConstant.COMMON_TAG_INDICARTOR, INDICATOR_RDB_LAST_BGSAVE_STATUS);
            point = Point.metric(metricInstanceMonitor).tag(tags).value(unixTime, infoAllVo.getRdbLastBgsaveStatus()).build(true);
            list.add(point);
        } catch (Exception ex) {
            LOGGER.error("InstanceInfoAllOpenTsdbBolt.processInfoInstanceMetric ----> 写入Opentsdb异常", ex);
        }
        return list;
    }

    /**
     * 入库命令监控大盘-基础监控
     *
     * @param infoAllVo
     */
    public static List<Point> processInfoCommandBaseMetric(InstanceInfoAllVo infoAllVo, String metricCommandBaseMonitor, long unixTime) {
        List<Point> list = new ArrayList<>();
        try {
            Map<String, String> tags = new HashMap<>(4);
            tags.put(OpentsdbConstant.COMMON_TAG_APPID, String.valueOf(infoAllVo.getAppId()));
            tags.put(OpentsdbConstant.COMMON_TAG_INSTANCE, infoAllVo.getIp() + "_" + infoAllVo.getPort());
            tags.put(OpentsdbConstant.COMMON_TAG_ROLE, String.valueOf(infoAllVo.getRole()));
            //入库键个数
            tags.put(OpentsdbConstant.COMMON_TAG_INDICARTOR, INDICATOR_KEY_COUNT);
            Point point = Point.metric(metricCommandBaseMonitor).tag(tags).value(unixTime, infoAllVo.getKeyCount()).build(true);
            list.add(point);
            //入库过期键个数
            tags.put(OpentsdbConstant.COMMON_TAG_INDICARTOR, INDICATOR_HAS_EXPIRED_KEY_COUNT);
            point = Point.metric(metricCommandBaseMonitor).tag(tags).value(unixTime, infoAllVo.getHasExpireKeyCount()).build(true);
            list.add(point);
            //入库已过期键个数（差值）
            tags.put(OpentsdbConstant.COMMON_TAG_INDICARTOR, INDICATOR_EXPIRED_KEY_COUNT_DIFF);
            point = Point.metric(metricCommandBaseMonitor).tag(tags).value(unixTime, infoAllVo.getExpiredKeysDiff()).build(true);
            list.add(point);
            //入库逐出键数量个数（差值）
            tags.put(OpentsdbConstant.COMMON_TAG_INDICARTOR, INDICATOR_EVICTED_KEY_COUNT_DIFF);
            point = Point.metric(metricCommandBaseMonitor).tag(tags).value(unixTime, infoAllVo.getEvictedKeysDiff()).build(true);
            list.add(point);
            //入库qps
            tags.put(OpentsdbConstant.COMMON_TAG_INDICARTOR, INDICATOR_QPS);
            point = Point.metric(metricCommandBaseMonitor).tag(tags).value(unixTime, infoAllVo.getQps()).build(true);
            list.add(point);
            //入库错误数量
            tags.put(OpentsdbConstant.COMMON_TAG_INDICARTOR, INDICATOR_MISSES_DIFF);
            point = Point.metric(metricCommandBaseMonitor).tag(tags).value(unixTime, infoAllVo.getMissesDiff()).build(true);
            list.add(point);
            //入库total_commands_processed (新增的)
            tags.put(OpentsdbConstant.COMMON_TAG_INDICARTOR, INDICATOR_TOTAL_COMMANDS_PROCESSED_DIFF);
            point = Point.metric(metricCommandBaseMonitor).tag(tags).value(unixTime, infoAllVo.getTotalCommandsProcessedDiff()).build(true);
            list.add(point);
        } catch (Exception ex) {
            LOGGER.error("InstanceInfoAllOpenTsdbBolt.processInfoCommandBaseMetric ----> 写入Opentsdb异常", ex);
        }
        return list;
    }

    /**
     * 入库集群统计/实例统计 相关信息
     *
     * @param infoAllVo
     */
    public static List<Point> processInfoAppInstanceMetric(InstanceInfoAllVo infoAllVo, String metricAppInstanceMonitor, long unixTime) {
        List<Point> list = new ArrayList<>();
        try {
            Map<String, String> tags = new HashMap<>(4);
            tags.put(OpentsdbConstant.COMMON_TAG_APPID, String.valueOf(infoAllVo.getAppId()));
            tags.put(OpentsdbConstant.COMMON_TAG_INSTANCE, infoAllVo.getIp() + "_" + infoAllVo.getPort());
            tags.put(OpentsdbConstant.COMMON_TAG_ROLE, String.valueOf(infoAllVo.getRole()));
            //入库qps
            tags.put(OpentsdbConstant.COMMON_TAG_INDICARTOR, INDICATOR_QPS);
            Point point = Point.metric(metricAppInstanceMonitor).tag(tags).value(unixTime, infoAllVo.getQps()).build(true);
            list.add(point);
            //入库对象数
            tags.put(OpentsdbConstant.COMMON_TAG_INDICARTOR, INDICATOR_KEY_COUNT);
            point = Point.metric(metricAppInstanceMonitor).tag(tags).value(unixTime, infoAllVo.getKeyCount()).build(true);
            list.add(point);
            //入库命中率
            tags.put(OpentsdbConstant.COMMON_TAG_INDICARTOR, INDICATOR_HIT_PERCENT);
            point = Point.metric(metricAppInstanceMonitor).tag(tags).value(unixTime, infoAllVo.getHitPercent()).build(true);
            list.add(point);
            //入库连接数
            tags.put(OpentsdbConstant.COMMON_TAG_INDICARTOR, INDICATOR_CURR_CONNECTIONS);
            point = Point.metric(metricAppInstanceMonitor).tag(tags).value(unixTime, infoAllVo.getCurrConnections()).build(true);
            list.add(point);
            //入库内存使用率
            tags.put(OpentsdbConstant.COMMON_TAG_INDICARTOR, INDICATOR_MEM_USED_PERCENT);
            point = Point.metric(metricAppInstanceMonitor).tag(tags).value(unixTime, infoAllVo.getMemUsePercent()).build(true);
            list.add(point);
            //入库内存使用量
            tags.put(OpentsdbConstant.COMMON_TAG_INDICARTOR, INDICATOR_MEM_USED);
            point = Point.metric(metricAppInstanceMonitor).tag(tags).value(unixTime, infoAllVo.getUsedMem()).build(true);
            list.add(point);
        } catch (Exception ex) {
            LOGGER.error("InstanceInfoAllOpenTsdbBolt.processInfoAppInstanceMetric ----> 写入Opentsdb异常", ex);
        }
        return list;
    }

    /**
     * metric
     *
     * @param infoCmdVos
     */
    public static List<Point> processCommandStatsToOpenTsdb(List<InstanceInfoCmdVo> infoCmdVos) {
        Map<String, String> tags = new HashMap<>(3);
        List<Point> list = new ArrayList<>();
        for (InstanceInfoCmdVo infoCmdVo : infoCmdVos) {
            tags.put(OpentsdbConstant.COMMON_TAG_APPID, String.valueOf(infoCmdVo.getAppId()));
            tags.put(OpentsdbConstant.COMMON_TAG_INSTID, String.valueOf(infoCmdVo.getInstanceId()));
            tags.put(OpentsdbConstant.COMMON_TAG_CMD_TYPE, infoCmdVo.getCommandType());
            try {
                list.add(Point.metric(OpentsdbConstant.METRIC_COMMAND_KEY_MONITOR).tag(tags).value(infoCmdVo.getCollectTime().getTime(), infoCmdVo.getCommandCount()).build(true));
            } catch (Exception ex) {
                LOGGER.error("写入OpenTsdb失败，appId:{},instId:{},commandType:{},collectTime:{}",
                        infoCmdVo.getAppId(), infoCmdVo.getInstanceId(), infoCmdVo.getCommandType(), infoCmdVo.getCollectTime().getTime());
            }
        }
        return list;
    }

    /**
     * 解析入库point
     *
     * @param infoAllVos
     * @return
     */
    public static List<Point> analyzeInfoToPoint(List<InstanceInfoAllVo> infoAllVos) {
        List<Point> records = new ArrayList<>();
        for (InstanceInfoAllVo infoAllVo : infoAllVos) {
            long unixTime = BambooUtils.getMillisTimeByUnixTime(infoAllVo.getCollectTime());
            List<Point> appInstanceLists = FlinkSupportor.processInfoAppInstanceMetric(infoAllVo, OpentsdbConstant.METRIC_APP_INSTANCE_MONITOR, unixTime);
            List<Point> cmdBaseLists = FlinkSupportor.processInfoCommandBaseMetric(infoAllVo, OpentsdbConstant.METRIC_COMMAND_BASE_MONITOR, unixTime);
            List<Point> infoInstanceLists = FlinkSupportor.processInfoInstanceMetric(infoAllVo, OpentsdbConstant.METRIC_INSTANCE_MONITOR, unixTime);
            records.addAll(appInstanceLists);
            records.addAll(cmdBaseLists);
            records.addAll(infoInstanceLists);
        }
        return records;
    }

    /**
     * 解析info commandstats 和 计算差值
     *
     * @param infoAllVo
     * @param oldInfoAllVo
     */
    public static List<InstanceInfoCmdVo> infoAllCommandStatsInternalCalculate(InstanceInfoAllVo infoAllVo, InstanceInfoAllVo oldInfoAllVo) {
        List<InstanceInfoCmdVo> infoCmdVos = new ArrayList<>();
        Map<String, Object> newCmdInfo = infoAllVo.getCmdInfos();
        Map<String, Object> oldCmdInfo = null;
        if (oldInfoAllVo != null) {
            oldCmdInfo = oldInfoAllVo.getCmdInfos();
        }
        Map<String, InnerClass> newCmd = calculateCmdInfoMap(newCmdInfo);
        Map<String, InnerClass> oldCmd = null;
        if (MapUtils.isNotEmpty(oldCmdInfo)) {
            oldCmd = calculateCmdInfoMap(oldCmdInfo);
        }
        if (MapUtils.isEmpty(newCmd)) {
            return infoCmdVos;
        }
        for (Map.Entry<String, InnerClass> entry : newCmd.entrySet()) {
            String key = entry.getKey();
            InnerClass newInnerClazz = entry.getValue();

            InstanceInfoCmdVo instanceInfoCmdVo = new InstanceInfoCmdVo();
            instanceInfoCmdVo.setAppId(infoAllVo.getAppId());
            instanceInfoCmdVo.setInstanceId(infoAllVo.getInstanceId());
            instanceInfoCmdVo.setCollectTime(BambooUtils.getDateFromUnixTime(infoAllVo.getCollectTime()));
            instanceInfoCmdVo.setCommandName(key);
            CommandMonitorEnum commandMonitorEnum = CommandMonitorEnum.getCommandMonitorEnum(key);
            instanceInfoCmdVo.setCommandType(commandMonitorEnum == null ? "other" : commandMonitorEnum.name().toLowerCase());
            if (MapUtils.isNotEmpty(oldCmd)) {
                InnerClass oldInnerClazz = oldCmd.get(key);
                if (oldInnerClazz != null) {
                    long diff = newInnerClazz.getCommandCount() - oldInnerClazz.getCommandCount();
                    if (diff > 0) {
                        instanceInfoCmdVo.setCommandCount(diff);
                        instanceInfoCmdVo.setCommandUsec(newInnerClazz.getCommandUsec() - oldInnerClazz.getCommandUsec());
                    }
                }
            }
            infoCmdVos.add(instanceInfoCmdVo);
        }
        return infoCmdVos;
    }

    /**
     * 计算使用率，计算差值
     *
     * @param infoAllVo
     */
    public static InstanceInfoAllVo infoAllVoInternalCalculate(InstanceInfoAllVo infoAllVo, InstanceInfoAllVo oldInfoAllVo) {
        //1、memUsePercent
        if (infoAllVo.getMaxMem() != null && infoAllVo.getMaxMem() != 0) {
            double result = infoAllVo.getUsedMem() / (infoAllVo.getMaxMem() * 1.0);
            result = BambooUtils.formatDouble2(result);
            infoAllVo.setMemUsePercent(String.valueOf(result));
        }
        //2、expiredKeysDiff
        if (oldInfoAllVo != null && infoAllVo.getExpiredKeyCount() > oldInfoAllVo.getExpiredKeyCount()) {
            infoAllVo.setExpiredKeysDiff(infoAllVo.getExpiredKeyCount() - oldInfoAllVo.getExpiredKeyCount());
        } else {
            infoAllVo.setExpiredKeysDiff(0L);
        }
        //3、evictedKeysDiff
        if (oldInfoAllVo != null && infoAllVo.getEvictedKeyCount() > oldInfoAllVo.getEvictedKeyCount()) {
            infoAllVo.setEvictedKeysDiff(infoAllVo.getEvictedKeyCount() - oldInfoAllVo.getEvictedKeyCount());
        } else {
            infoAllVo.setEvictedKeysDiff(0L);
        }
        //4、missesDiff
        if (oldInfoAllVo != null && infoAllVo.getMisses() > oldInfoAllVo.getMisses()) {
            infoAllVo.setMissesDiff(infoAllVo.getMisses() - oldInfoAllVo.getMisses());
        } else {
            infoAllVo.setMissesDiff(0L);
        }
        //5、hitsDiff
        if (oldInfoAllVo != null && infoAllVo.getHits() > oldInfoAllVo.getHits()) {
            infoAllVo.setHitsDiff(infoAllVo.getHits() - oldInfoAllVo.getHits());
        } else {
            infoAllVo.setHitsDiff(0L);
        }
        //6、hitPercent
        if (infoAllVo.getHitsDiff() + infoAllVo.getMissesDiff() > 0) {
            double result = infoAllVo.getHits() / ((infoAllVo.getHits() + infoAllVo.getMisses()) * 1.0);
            result = BambooUtils.formatDouble2(result);
            infoAllVo.setHitPercent(String.valueOf(result));
        }
        //7、totalCommandsProcessedDiff
        if (oldInfoAllVo != null && infoAllVo.getTotalCommandsProcessed() > oldInfoAllVo.getTotalCommandsProcessed()) {
            infoAllVo.setTotalCommandsProcessedDiff((int) (infoAllVo.getTotalCommandsProcessed() - oldInfoAllVo.getTotalCommandsProcessed()));
        } else {
            infoAllVo.setTotalCommandsProcessedDiff(0);
        }
        return infoAllVo;
    }

    /**
     * 解析Map
     * cmdstat_get:calls=690942,usec=5037172,usec_per_call=7.29
     *
     * @param newCmdInfo
     * @return
     */
    public static Map<String, InnerClass> calculateCmdInfoMap(Map<String, Object> newCmdInfo) {
        if (MapUtils.isEmpty(newCmdInfo)) {
            return null;
        }
        Map<String, InnerClass> map = new HashMap<>(16);
        for (Map.Entry<String, Object> entry : newCmdInfo.entrySet()) {
            String[] keys = entry.getKey().split("_");
            List<String> vals = Splitter.on(CharMatcher.anyOf("=,")).omitEmptyStrings().trimResults().splitToList((String) entry.getValue());
            map.put(keys[1], new InnerClass(Long.parseLong(vals.get(1)), Long.parseLong(vals.get(3))));
        }
        return map;
    }

    /**
     * 获取最大的两个元素
     *
     * @param list
     * @param top
     * @return
     */
    public static List<InstanceInfoAllVo> listOrder(List<InstanceInfoAllVo> list, int top) {
        return Ordering.natural().onResultOf(new Function<InstanceInfoAllVo, Long>() {
            @Nullable
            @Override
            public Long apply(@Nullable InstanceInfoAllVo input) {
                assert input != null;
                return input.getCollectTime();
            }
        }).greatestOf(list, 2);
    }

    public static void main(String[] args) {
        List<InstanceInfoAllVo> list = new ArrayList<>();
        InstanceInfoAllVo vo1 = new InstanceInfoAllVo();
        vo1.setCollectTime(111L);
        InstanceInfoAllVo vo2 = new InstanceInfoAllVo();
        vo2.setCollectTime(222L);
        InstanceInfoAllVo vo3 = new InstanceInfoAllVo();
        vo3.setCollectTime(333L);
        list.add(vo1);
        list.add(vo2);
        list.add(vo3);
        list = listOrder(list, 2);
        System.out.print("第一个元素：" + list.get(0).getCollectTime());
        System.out.print("第二个元素：" + list.get(1).getCollectTime());

    }
}
